package com.hsgene.util.mq.rocket.annotation;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.google.common.collect.Maps;
import com.hsgene.util.mq.rocket.config.RocketMQConfigurationProperties;
import com.hsgene.util.mq.rocket.consumer.AbstractRocketMQListener;
import com.hsgene.util.mq.rocket.consumer.RocketMQTopicListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.Map;
import java.util.Properties;

/**
 * 此处在容器初始化完成后注册RocketMQ消息监听
 *
 * @author wangbing
 * @version 1.0
 * @date 2019/1/16 17:29
 * Copyright: Copyright (c) 2019
 */
@Component
public class RocketMQApplicationContext implements ApplicationContextAware {

    private final static Logger LOGGER = LoggerFactory.getLogger(RocketMQApplicationContext.class);

    @Resource
    private RocketMQConfigurationProperties properties;

    private Properties consumerProperties;

    private Map<String, Map<String, AbstractRocketMQListener>> topicTagListenerMap = Maps.newHashMap();

    public void init() {
        consumerProperties = new Properties();
        consumerProperties.put(PropertyKeyConst.ONSAddr, properties.getNamsrvAddr());
        consumerProperties.put(PropertyKeyConst.ConsumerId, properties.getConsumerId());
        consumerProperties.put(PropertyKeyConst.AccessKey, properties.getAccessKey());
        consumerProperties.put(PropertyKeyConst.SecretKey, properties.getSecretKey());
    }

    @Override
    public void setApplicationContext(ApplicationContext context) throws BeansException {
        Map<String, Object> listenerMap = context.getBeansWithAnnotation(RocketMQListener.class);
        if (!CollectionUtils.isEmpty(listenerMap)) {
            if (consumerProperties == null) {
                init();
            }
            listenerMap.forEach((key, obj) -> {
                if (obj instanceof AbstractRocketMQListener) {
                    AbstractRocketMQListener mqListener = (AbstractRocketMQListener) obj;
                    RocketMQListener listener = mqListener.getClass().getAnnotation(RocketMQListener.class);
                    String receiveTopic = listener.topic();
                    String receiveTag = listener.tag();
                    Object topic = properties.getTopic(listener.topic());
                    if (topic == null) {
                        throw new IllegalArgumentException("无法找到[" + receiveTopic + "]对应的Topic");
                    }
                    Object tag = properties.getTag(receiveTag);
                    if (tag == null) {
                        throw new IllegalArgumentException("无法找到[" + receiveTag + "]对应的Tag");
                    }
                    Map<String, AbstractRocketMQListener> tagListenerMap = topicTagListenerMap.get(topic.toString());
                    if (CollectionUtils.isEmpty(tagListenerMap)) {
                        tagListenerMap = Maps.newHashMap();
                        topicTagListenerMap.put(topic.toString(), tagListenerMap);
                    }
                    tagListenerMap.put(tag.toString(), mqListener);
                } else {
                    // @RocketMQListener不是注解的实现了AbstractRocketMQListener的类，对其不做处理
                }
            });
        }
        if (!CollectionUtils.isEmpty(topicTagListenerMap)) {
            topicTagListenerMap.forEach((topic, tagListenerMap) -> {
                Consumer consumer = ONSFactory.createConsumer(consumerProperties);
                RocketMQTopicListener topicListener = new RocketMQTopicListener(tagListenerMap);
                // 主题式消息监听，消费者监听主题下所有消息，保持消费者集群订阅关系一致
                consumer.subscribe(topic, "*", topicListener);
                consumer.start();
                tagListenerMap.forEach((tag, listener) ->
                        LOGGER.info("{}-->正在监听Topic:{}, Tag:{}", listener.getClass().getName(), topic, tag));
            });
        } else {
            LOGGER.info("当前没有需要监听的内容");
        }
    }
}
